package com.suxin.queue.batch;

import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.springframework.util.CollectionUtils;

import javax.annotation.PostConstruct;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;

/**
 * @author Tang
 * @classname AbstractLazyPending
 * @description [ 延迟消费：生产者-消费者模型 ]
 * @date 2022/2/16 17:38
 */
@Data
@Slf4j
public abstract class AbstractLazyPending<E> {

    /**
     * 需要子类去创建
     */
    protected PendingParam<E> pendingParam;

    /**
     * 带处理的任务
     */
    private List<E> tasks = new ArrayList<>();

    /**
     * 最后处理数据的时间
     */
    private Long lastHandlerTime = System.currentTimeMillis();

    /**
     * 单线程消费
     */
    @PostConstruct
    public void initConsumerPending() {
        Thread thread = new Thread(() -> {
            while (true) {
                try {
                    E obj = this.pendingParam.getQueue()
                            .poll(this.pendingParam.getTimeThreshold(), TimeUnit.MILLISECONDS);
                    if (null != obj) {
                        tasks.add(obj);
                    }
                    // 1.数量超限 2.超时
                    if (!CollectionUtils.isEmpty(tasks) && dataReady()) {
                        lastHandlerTime = System.currentTimeMillis();
                        List<E> taskRef = tasks;
                        tasks = new ArrayList<>();
                        // 批处理
                        this.pendingParam.getExecutorService().execute(() -> {
                            doHandler(taskRef);
                        });
                    }
                } catch (InterruptedException e) {
                    log.error("Pending#initConsumePending failed:{}",e.getMessage());
                }
            }
        });
        thread.start();

    }

    /**
     * doHandler 具体的业务逻辑
     * @param taskRef
     */
    protected abstract void doHandler(List<E> taskRef);

    /**
     * dataReady
     * @return
     */
    private boolean dataReady() {
        return (tasks.size() >= this.pendingParam.getNumThreshold()) ||
                (System.currentTimeMillis() - lastHandlerTime >= this.pendingParam.getTimeThreshold());
    }

    /**
     * pending
     * @param e
     */
    private void pending(E e) {
        try {
            this.pendingParam.getQueue().put(e);
        } catch (InterruptedException interruptedException) {
            interruptedException.printStackTrace();
        }
    }

}